package com.zx.learn.flink.source;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 使用env.generateSequence()方法创建基于Sequence的DataStream
 * 使用env.fromSequence()方法创建基于开始和结束的DataStream
 */
public class FromParCollectionDemo {
    public static void main(String[] args) throws Exception {
        //直接获取当前的运行环境
        //StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port", 8081);//设置webui的端口号
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);

        //设置全局的并行度
        env.setParallelism(10);
        /**
         * nums创建的DataStream的并行度为：16
         * nums2创建的DataStream的并行度为：2
         * nums3创建的DataStream的并行度为：16
         */
        DataStreamSource<Long> nums = env.generateSequence(1, 10);
        DataStreamSource<Long> nums2 = env.generateSequence(1, 10).setParallelism(2);
        DataStreamSource<Long> nums3 = env.fromSequence(1, 10);

        System.out.println("nums创建的DataStream的并行度为："+nums.getParallelism());
        System.out.println("nums2创建的DataStream的并行度为："+nums2.getParallelism());
        System.out.println("nums3创建的DataStream的并行度为："+nums3.getParallelism());
        nums.print();
        nums2.print();
        nums3.print();
        env.execute();
    }
}
